package com.ebupt.migu.music.schedule.service.Impl;

import cn.hutool.core.io.file.FileReader;
import cn.hutool.core.io.file.FileWriter;
import com.ebupt.migu.common.pojo.ErrorReport;
import com.ebupt.migu.common.pojo.UserGroupStatus;
import com.ebupt.migu.common.util.RedisLock;
import com.ebupt.migu.music.common.constants.StrategyConstants;
import com.ebupt.migu.music.common.util.DateUtil;
import com.ebupt.migu.music.common.util.FileRecordAsync;
import com.ebupt.migu.music.common.util.HttpUtils;
import com.ebupt.migu.music.common.util.ListUtil;
import com.ebupt.migu.music.schedule.entity.SendMessageEntity;
import com.ebupt.migu.music.schedule.mapper.MessageNoticeTaskMapper;
import com.ebupt.migu.music.schedule.service.MessageNoticeService;
import com.ebupt.migu.music.schedule.service.MessageNoticeTaskService;
import com.ebupt.migu.restapi.DownloadByFtp;
import com.ebupt.migu.restapi.GroupDetailDel;
import com.ebupt.migu.restapi.IopMatchPost;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.data.redis.support.atomic.RedisAtomicLong;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.io.BufferedWriter;
import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * Created by Intellij IDEA. User: liuyangyang@cd.ebupt.com Date: 2020/3/21
 */
@Slf4j
@Service
public class MessageNoticeTaskServiceImpl implements MessageNoticeTaskService {

    @Autowired
    private MessageNoticeTaskMapper messageNoticeTaskMapper;
    @Resource
    private RedisTemplate redisTemplate;
    @Autowired
    private MessageNoticeService messageNoticeService;
    @Autowired
    private FileRecordAsync fileRecordAsync;
    @Value("${message.send_url}")
    private String SEND_MESSAGE_URL;
    @Value("${label.post_url}")
    private String postUrl;

    /**
     * 消息通知策略： 立即发送消息通知
     */
    @Override
    public void sendMessageNow() {
        Date date = new Date();
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat(DateUtil.DATE_TIME_FORMAT_YYYY_MM_DD_HH_MI_SS);
        // 当前时间
        String now = simpleDateFormat.format(date);
        log.info("当前时间:{}", now);
        // 查询需要立即进行消息通知的策略列表
        List<HashMap<String, Object>> strategies = messageNoticeTaskMapper.querySendMessageNow();
        log.info("需要立即进行消息通知的策略共有:{}条", strategies.size());
        if (null == strategies || strategies.isEmpty()) {
            log.info("当前不存在需要立即进行消息通知的策略");
            return;
        }
        long start = System.currentTimeMillis();
        Map<String, Object> strategy = strategies.get(0);
        // 获取策略ID更新相应状态
        String strategyId = strategy.get("strategyId").toString();
        // 生成task id
        SimpleDateFormat simpleDateFormat1 = new SimpleDateFormat(DateUtil.DATE_FORMAT_YYYYMMDD);
        String taskDate = simpleDateFormat1.format(date);
        String taskID = strategyId + taskDate;
        // 用户群ID
        String groupId = strategy.get("groupId").toString();
        // 根据策略关联人群状态判断是否执行下发操作
        if (isSend(strategy, now)) {
            RedisLock redisLock = new RedisLock(redisTemplate, strategyId);
            if (!redisLock.getLock()) {
                log.info("{}已在其他调度中开始下发，本机终止运营", strategyId);
                return;
            }
            log.info("执行策略Id:{},策略名称:{},taskId:{},groupId:{}", strategyId, strategy.get("strategyName").toString(),
                    taskID, groupId);
            // 创建sftp任务
            String recordFile = create_sftp_data_sync_record(groupId, strategyId);
            // 分支流程
            pro(groupId, strategyId, strategy, taskID, 1, recordFile);
            createMSGCallBack(strategyId, taskDate);
        } else {
            log.info("未执行的策略Id:{},名称:{},人群状态异常", strategyId, strategy.get("strategyName").toString());
        }
        // 将策略状态修改为已完成
        strategy_finished(strategyId, 1);
        log.info("本次消息通知耗时{}ms", System.currentTimeMillis() - start);
    }

    /**
     * 消息通知策略： 单次发送消息通知
     */
    @Override
    public void sendMessageOnce() {
        Date date = new Date();
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat(DateUtil.DATE_TIME_FORMAT_YYYY_MM_DD_HH_MI_SS);
        // 当前时间
        String now = simpleDateFormat.format(date);
        // 查询需要立即发送的消息通知策略列表
        List<HashMap<String, Object>> strategies = messageNoticeTaskMapper.querySendMessageOnce(now);
        if (null == strategies || strategies.isEmpty()) {
            log.info("当前不存在单次发送消息通知的策略");
            return;
        }
        long start = System.currentTimeMillis();
        Map<String, Object> strategy = strategies.get(0);
        // 获取策略ID更新相应状态
        String strategyId = strategy.get("strategyId").toString();
        // 生成task id
        SimpleDateFormat simpleDateFormat1 = new SimpleDateFormat(DateUtil.DATE_FORMAT_YYYYMMDD);
        String taskDate = simpleDateFormat1.format(date);
        String taskID = strategyId + taskDate;
        // 用户群ID
        String groupId = strategy.get("groupId").toString();
        // 根据策略关联人群状态判断是否执行下发操作
        if (isSend(strategy, now)) {
            RedisLock redisLock = new RedisLock(redisTemplate, strategyId);
            if (!redisLock.getLock()) {
                log.info("{}已在其他调度中开始下发，本机终止运营", strategyId);
                return;
            }
            log.info("执行策略Id:{},策略名称:{},taskId:{},groupId:{}", strategyId, strategy.get("strategyName").toString(),
                    taskID, groupId);
            // 创建sftp任务
            String recordFile = create_sftp_data_sync_record(groupId, strategyId);
            // 分支流程
            pro(groupId, strategyId, strategy, taskID, 1, recordFile);
            createMSGCallBack(strategyId, taskDate);
        } else {
            log.info("未执行的策略Id:{},名称:{},人群状态异常", strategyId, strategy.get("strategyName").toString());
        }
        // 将策略状态修改为已完成
        strategy_finished(strategyId, 1);
        log.info("redis中获取目标人群到策略执行完成耗时:{}", System.currentTimeMillis() - start);
    }

    /**
     * 消息通知策略： 循环发送消息通知
     */
    @Override
    public void sendMessageLoop() throws ParseException {
        Date date = new Date();
        // 年月日时分秒
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat(DateUtil.DATE_TIME_FORMAT_YYYY_MM_DD_HH_MI_SS);
        // 年月日
        SimpleDateFormat simpleDateFormat_day = new SimpleDateFormat(DateUtil.DATE_FORMAT_YYYY_MM_DD);
        // 时分秒
        SimpleDateFormat simpleDateFormat_time = new SimpleDateFormat(DateUtil.DATE_TIME_FORMAT_HH_MI_SS);
        // 当前年月日时分秒
        String now = simpleDateFormat.format(date);
        // 当前年月日
        String dayOnly = simpleDateFormat_day.format(date);
        // 当前时分秒
        String timeOnly = simpleDateFormat_time.format(date);

        // 查询需要循环发送消息通知的策略列表
        List<HashMap<String, Object>> strategies = messageNoticeTaskMapper.querySendMessageLoop(dayOnly);
        if (null == strategies || strategies.isEmpty()) {
            log.info("当前不存在循环发送消息通知的策略");
            return;
        }
        Map<String, Object> strategy = new HashMap<>();
        for (Map<String, Object> s : strategies) {
            String startTime = s.get("startTime").toString();
            String[] starts = startTime.split(" ");
            String start = starts[1];
            // 最近一次执行时间为空且执行时间小于当前时间
            if (s.get("executeTime") == null
                    && (simpleDateFormat_time.parse(start).getTime() <= simpleDateFormat_time.parse(timeOnly).getTime())) {
                strategy = s;
                break;
            }
            if (s.get("executeTime") != null) {
                String executeTime = s.get("executeTime").toString();
                String[] executes = executeTime.split(" ");
                String execute = executes[0];
                // 最近一次执行时间不为空且最后一次执行日期小于今天且执行时间小于当前时间
                if ((simpleDateFormat_day.parse(execute).getTime() < simpleDateFormat_day.parse(dayOnly).getTime())
                        && (simpleDateFormat_time.parse(start).getTime() <= simpleDateFormat_time.parse(timeOnly)
                        .getTime())) {
                    strategy = s;
                    break;
                }
            }
        }
        if (null == strategy.get("strategyName")) {
            log.info("当前不存在循环发送消息通知的策略");
            return;
        }
        long start = System.currentTimeMillis();
        // 获取策略ID更新相应状态
        String strategyId = strategy.get("strategyId").toString();
        // 生成task id
        SimpleDateFormat simpleDateFormat1 = new SimpleDateFormat(DateUtil.DATE_FORMAT_YYYYMMDD);
        String taskDate = simpleDateFormat1.format(date);
        String taskID = strategyId + taskDate;
        // 用户群ID
        String groupId = strategy.get("groupId").toString();
        // 根据策略关联人群状态判断是否执行下发操作
        if (isSend(strategy, now)) {
            RedisLock redisLock = new RedisLock(redisTemplate, strategyId);
            if (!redisLock.getLock()) {
                log.info("{}已在其他调度中开始下发，本机终止运营", strategyId);
                return;
            }
            log.info("执行策略Id:{},策略名称:{},taskId:{},groupId:{}", strategyId, strategy.get("strategyName").toString(),
                    taskID, groupId);
            // 创建sftp任务
            String recordFile = create_sftp_data_sync_record(groupId, strategyId);
            // 分支流程
            pro(groupId, strategyId, strategy, taskID, 2, recordFile);
            createMSGCallBack(strategyId, taskDate);
        } else {
            log.info("未执行的策略Id:{},名称:{},人群状态异常", strategyId, strategy.get("strategyName").toString());
        }
        // 策略截止日期
        String endTime = strategy.get("endTime").toString();
        Map<String, Object> map = new HashMap<>();
        map.put("strategyId", strategyId);
        map.put("executing", 1);
        // 截止日期是今天，将策略状态修改成已完成。 不是今天保留进行中
        String[] ends = endTime.split(" ");
        String end = ends[0];
        if (end.equals(dayOnly)) {
            map.put("executeStatus", StrategyConstants.EXECUTE_ENDED);
            log.info("将策略状态修改为已完成");
        }
        messageNoticeTaskMapper.updateStrategyStatus(map);
        log.info("redis中获取目标人群到策略执行完成耗时:{}", System.currentTimeMillis() - start);
    }

    /**
     * 消息通知策略： 间隔发送消息通知
     *
     * @throws ParseException
     */
    @Override
    public void sendMessageInterval() throws ParseException {
        Date date = new Date();
        // 年月日时分秒
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat(DateUtil.DATE_TIME_FORMAT_YYYY_MM_DD_HH_MI_SS);
        // 年月日
        SimpleDateFormat simpleDateFormat_day = new SimpleDateFormat(DateUtil.DATE_FORMAT_YYYY_MM_DD);
        // 时分秒
        SimpleDateFormat simpleDateFormat_time = new SimpleDateFormat(DateUtil.DATE_TIME_FORMAT_HH_MI_SS);
        // 当前年月日时分秒
        String now = simpleDateFormat.format(date);
        // 当前年月日
        String dayOnly = simpleDateFormat_day.format(date);
        // 当前时分秒
        String timeOnly = simpleDateFormat_time.format(date);

        // 查询需要循环发送消息通知的策略列表
        List<HashMap<String, Object>> strategies = messageNoticeTaskMapper.querySendMessageInterval(dayOnly);
        if (null == strategies || strategies.isEmpty()) {
            log.info("当前不存在间隔发送消息通知的策略");
            return;
        }
        Map<String, Object> strategy = new HashMap<>();
        for (Map<String, Object> s : strategies) {
            String startTime = s.get("startTime").toString();
            String[] starts = startTime.split(" ");
            String start = starts[1];
            // 间隔时间
            int intervalDays = Integer.parseInt(s.get("intervalDays").toString()) + 1;
            // 最近一次执行时间为空且执行时间小于当前时间
            if (s.get("executeTime") == null
                    && (simpleDateFormat_time.parse(start).getTime() <= simpleDateFormat_time.parse(timeOnly).getTime())) {
                strategy = s;
                break;
            }
            if (s.get("executeTime") != null) {
                String executeTime = s.get("executeTime").toString();
                String[] executes = executeTime.split(" ");
                String execute = executes[0];
                // 最近一次执行时间不为空且最后一次执行日期小于今天且执行时间小于当前时间
                if ((simpleDateFormat_day.parse(execute).getTime()
                        + intervalDays * 24 * 60 * 60 * 1000 == simpleDateFormat_day.parse(dayOnly).getTime())
                        && (simpleDateFormat_time.parse(start).getTime() <= simpleDateFormat_time.parse(timeOnly)
                        .getTime())) {
                    strategy = s;
                    break;
                }
            }
        }

        if (null == strategy.get("strategyName")) {
            log.info("当前不存在循环发送消息通知的策略");
            return;
        }
        long start = System.currentTimeMillis();
        // 获取策略ID更新相应状态
        String strategyId = strategy.get("strategyId").toString();
        // 生成task id
        SimpleDateFormat simpleDateFormat1 = new SimpleDateFormat(DateUtil.DATE_FORMAT_YYYYMMDD);
        String taskDate = simpleDateFormat1.format(date);
        String taskID = strategyId + taskDate;
        // 用户群ID
        String groupId = strategy.get("groupId").toString();
        // 根据策略关联人群状态判断是否执行下发操作
        if (isSend(strategy, now)) {
            RedisLock redisLock = new RedisLock(redisTemplate, strategyId);
            if (!redisLock.getLock()) {
                log.info("{}已在其他调度中开始下发，本机终止运营", strategyId);
                return;
            }
            log.info("执行策略Id:{},策略名称:{},taskId:{},groupId:{}", strategyId, strategy.get("strategyName").toString(),
                    taskID, groupId);
            // 创建sftp任务
            String recordFile = create_sftp_data_sync_record(groupId, strategyId);
            // 分支流程
            pro(groupId, strategyId, strategy, taskID, 2, recordFile);
            createMSGCallBack(strategyId, taskDate);
        } else {
            log.info("未执行的策略Id:{},名称:{},人群状态异常", strategyId, strategy.get("strategyName").toString());
        }
        // 策略截止日期
        String endTime = strategy.get("endTime").toString();
        Map<String, Object> map = new HashMap<>();
        map.put("strategyId", strategyId);
        map.put("executing", 1);
        // 截止日期是今天，将策略状态修改成已完成。 或下一次执行时间超出截止时间，将策略状态修改为已完成
        int intervalDays = Integer.parseInt(strategy.get("intervalDays").toString()) + 1;
        String[] ends = endTime.split(" ");
        String end = ends[0];
        if ((simpleDateFormat_day.parse(dayOnly).getTime() + intervalDays * 24 * 60 * 60 * 1000 > simpleDateFormat_day
                .parse(end).getTime()) || (end.equals(dayOnly))) {
            map.put("executeStatus", StrategyConstants.EXECUTE_ENDED);
            log.info("将策略状态修改为已完成");
        }
        messageNoticeTaskMapper.updateStrategyStatus(map);
        log.info("redis中获取目标人群到策略执行完成耗时:{}", System.currentTimeMillis() - start);
    }

    /**
     * 流程分支
     */
    private void pro(String groupId, String strategyId, Map<String, Object> strategy, String taskID, int type,
                     String recordFile) {
        String key_black = strategy.get("strategyId") + "_" + DateUtil.getDateToday() + "_black";
        String key_iop = strategy.get("strategyId") + "_" + DateUtil.getDateToday() + "_iop";
        // 发送人群集合
        List<String> uids;
        // 获取黑白名单成员
        Map<String, List<String>> members = blackOrWhite(strategy);
        // 判断用户群是否属于标签服务平台
        HashMap<String, Object> labelUserGroup = messageNoticeTaskMapper.isLabelUserGroup(groupId);
        // 创建临时KEY
        String redisKey = groupId + "-uid";

        if (!redisTemplate.hasKey(redisKey) && null != labelUserGroup && labelUserGroup.size() > 0) {
//             if (null != labelUserGroup && labelUserGroup.size() > 0) {
            // 对接标签服务平台取人群
            File file = DownloadByFtp.getLabelUserGroup(strategy, postUrl, 3);
            if (null == file || file.length() == 0) {
                log.info("标签客群文件为空");
                return;
            }
            log.info("成功获取到客群文件");
            List<String> list = new FileReader(file).readLines();
            uids = uidList(strategy, list, members);
            if (!CollectionUtils.isEmpty(uids)) {
                // 调用发送方法
                log.info("发送uid集合：{}", uids.toString());
                send(uids, strategy, taskID);
                // 明细记录
                try {
                    BufferedWriter writer = new FileWriter(recordFile).getWriter(true);
                    for (String uid : uids) {
                        // 明细记录
                        writer.write(uid + "|^|" + strategyId);
                        writer.newLine();
                    }
                    writer.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            // 对接标签平台客群销毁通知
            if (file.delete()) {
                GroupDetailDel.del(strategy);
            }
        } else {
            // 判断目标人群是否存在
            redis_set_size(redisKey, strategyId, type);
            // 遍历 -从redis中获取目标人群uid
            Cursor<String> cursor = getRedisCursor(redisTemplate, redisKey, 0);
            uids = uidList(strategy, cursor, members);
            if (!CollectionUtils.isEmpty(uids)) {
                // 调用发送方法
                send(uids, strategy, taskID);
            }
        }
        // 黑名单
        redisTemplate.expire(key_black, 30, TimeUnit.DAYS);

        // iop
        redisTemplate.expire(key_iop, 30, TimeUnit.DAYS);

    }

    private Cursor<String> getRedisCursor(RedisTemplate redisTemplate, String key, Integer num) {
        try {
            Cursor scan = redisTemplate.opsForSet().scan(key, ScanOptions.NONE);
            return scan;
        } catch (Exception e) {
            e.printStackTrace();
            try {
                Thread.sleep(3000L);
            } catch (Exception es) {
                log.info("重调获取内容间歇异常" + key);
            }
            int count = num + 1;
            log.info("次数重调获取内容：" + count + ",key:" + key);
            return getRedisCursor(redisTemplate, key, count);
        }
    }

    @Override
    public long groupSize(String groupId) {
        // 获取目标用户规模
        long size = redisTemplate.opsForSet().size(groupId);
        log.info("redisKey:{},获取size:{}", groupId, size);
        return size;
    }

    /**
     * 判断目标人群大小
     *
     * @param redis_key
     * @param sid
     * @param type      1- 已完成；
     */
    private void redis_set_size(String redis_key, String sid, int type) {
        // 获取目标用户规模
        Long start = System.currentTimeMillis();
        long size = redisTemplate.opsForSet().size(redis_key);
        if (size == 0) {
            log.info("目标人群不存在，无法进行消息推送:{}", redis_key);
            strategy_finished(sid, type);
            // 给人群创建人和维护人发送消息
            HttpUtils.doPost(SEND_MESSAGE_URL, new SendMessageEntity("1", "1", null, redis_key.replace("-uid", "")));
            HttpUtils.doPost(SEND_MESSAGE_URL, new SendMessageEntity("2", "1", null, redis_key.replace("-uid", "")));
            return;
        }
        log.info("目标人群共有:{}用户,耗时:{}ms", size, System.currentTimeMillis() - start);
    }

    /**
     * 根据策略关联人群状态判断是否执行下发操作
     *
     * @param strategy
     * @param time
     * @return
     */
    private boolean isSend(Map<String, Object> strategy, String time) {
        boolean isSend;
        String strategyId = strategy.get("strategyId").toString();
        String groupId = strategy.get("groupId").toString();
        String status = messageNoticeTaskMapper.qryUserGroupStatus(groupId);
        Map<String, Object> map = new HashMap<>();
        map.put("strategyId", strategyId);
        map.put("executeTime", time);
        if (UserGroupStatus.FINISH.getVal().equals(status)) {
            map.put("executeStatus", StrategyConstants.EXECUTE_ONGOING);
            map.put("executing", 2);
            isSend = true;
        } else {
            // 人群不是已完成状态，不能发送，记录下发失败日志
            ErrorReport errorReport = new ErrorReport();
            errorReport.setGid(groupId).setSid(strategyId).setCreateTime(time)
                    .setRemark(UserGroupStatus.getMsg(status));
            messageNoticeTaskMapper.insertErrorReport(errorReport);
            isSend = false;
        }
        // 更新
        messageNoticeTaskMapper.updateStrategyStatus(map);
        return isSend;
    }

    /**
     * 将策略状态修改为已完成
     *
     * @param sid
     * @param type 1- 已完成； 2- 进行中；
     */
    private void strategy_finished(String sid, int type) {
        Map<String, Object> map = new HashMap<>();
        map.put("strategyId", sid);
        map.put("executing", 1);
        if (type == 1) {
            map.put("executeStatus", StrategyConstants.EXECUTE_ENDED);
        }
        log.info("将策略状态修改为已完成");
        messageNoticeTaskMapper.updateStrategyStatus(map);
    }

    /**
     * 判断是否需要添加白名单或者剔除黑名单
     *
     * @param strategy
     */
    private Map<String, List<String>> blackOrWhite(Map<String, Object> strategy) {
        Long start = System.currentTimeMillis();
        Map<String, List<String>> map = new HashMap<>();

        // 白名单： 1-添加白名单
        String[] white = null;
        // 黑名单： 1-剔除黑名单
        String[] black = null;

        // 白名单成员
        List<String> whites = new ArrayList<>();
        // 黑名单成员
        List<String> blacks = new ArrayList<>();
        if (null != strategy.get("white")) {
            // 多选白名单
            white = String.valueOf(strategy.get("white")).split(",");
        }
        if (null != strategy.get("black")) {
            // 多选黑名单
            black = String.valueOf(strategy.get("black")).split(",");
        }
        // 添加白名单用户
        if (white != null && white.length > 0) {
            log.info("添加白名单用户:{}", white);
            whites = addWhiteList(white);
        }
        // 剔除黑名单用户
        if (black != null && black.length > 0) {
            log.info("剔除黑名单用户:{}", black);
            blacks = removeBlackList(black);
        }
        if (blacks != null && blacks.size() != 0) {
            map.put("blacks", blacks);
        }
        if (whites != null && whites.size() != 0) {
            map.put("whites", whites);
        }
        log.info("判断或筛选黑白名单耗时:{}ms", System.currentTimeMillis() - start);
        return map;
    }

    /**
     * 剔除黑名单用户
     */
    private List<String> removeBlackList(String[] ids) {
        List<String> members = new ArrayList<>();
        for (String id : ids) {
            List<String> blacks = messageNoticeTaskMapper.queryRosterMembers(id);
            if (null != blacks && !blacks.isEmpty()) {
                members.addAll(blacks);
            }
        }
        return members;
    }

    /**
     * 添加白名单用户
     */
    private List<String> addWhiteList(String[] ids) {
        List<String> members = new ArrayList<>();
        for (String id : ids) {
            List<String> whites = messageNoticeTaskMapper.queryRosterMembers(id);
            if (null != whites && !whites.isEmpty()) {
                members.addAll(whites);
            }
        }
        return members;
    }

    /**
     * 创建消息通知效果统计记录
     *
     * @param sid
     * @param date
     */
    private void createMSGCallBack(String sid, String date) {
        Map<String, Object> map = new HashMap<>();
        map.put("sid", sid);
        map.put("date", date);
        messageNoticeTaskMapper.createMSGCallback(map);
    }

    /**
     * 发送调用方法
     *
     * @param uids
     * @param taskID
     */
    private void send(List<String> uids, Map<String, Object> strategy, String taskID) {
        // iop验证
        if (StringUtils.isNotEmpty(strategy.get("iopTagId").toString())) {
            String key_iop = strategy.get("strategyId") + "_" + DateUtil.getDateToday() + "_iop";
            IopMatchPost ip = new IopMatchPost();
            uids = uids.stream()
                    .filter(
                            u -> ip.iopMatch(strategy.get("iopTagId").toString(), strategy.get("strategyId").toString(), "", u))
                    .collect(Collectors.toList());

            uids.forEach(s -> {
                redis_incr(key_iop);
            });

        }
        // 分组构造
        List<List<String>> list = ListUtil.spiltList(uids, 10);
        log.info("发送总数:{},分组总数:{}", uids.size(), list.size());
        list.forEach(strings -> {
            // 异步调用发送方法
            fileRecordAsync.record(strings, strategy, taskID);
            // log.info("模拟下发过程：{}", strings);
        });
    }

    /**
     * 获取全部的uid集合
     *
     * @param cursor
     * @param members
     * @return
     */
    private List<String> uidList(Map<String, Object> strategy, Cursor<String> cursor,
                                 Map<String, List<String>> members) {
        String key_black = strategy.get("strategyId") + "_" + DateUtil.getDateToday() + "_black";
        List<String> whites = new ArrayList<>();
        List<String> blacks = new ArrayList<>();
        if (members != null && members.get("whites") != null) {
            whites = members.get("whites");
        }
        if (members != null && members.get("blacks") != null) {
            blacks = members.get("blacks");
        }
        List<String> uids = new ArrayList<>();
        String uid;
        while (cursor.hasNext()) {
            uid = cursor.next();
            if (blacks != null && blacks.contains(uid)) {
                // 黑名单用户
                redis_incr(key_black);
                continue;
            }
            if (whites != null && whites.contains(uid)) {
                continue;
            }
            uids.add(uid);
        }
        if (whites != null && whites.size() != 0) {
            uids.addAll(whites);
        }
        return uids;
    }

    /**
     * 获取全部的uid集合
     *
     * @param strategy
     * @param list
     * @param members
     * @return
     */
    private List<String> uidList(Map<String, Object> strategy, List<String> list, Map<String, List<String>> members) {
        String key_black = strategy.get("strategyId") + "_" + DateUtil.getDateToday() + "_black";
        List<String> whites = new ArrayList<>();
        List<String> blacks = new ArrayList<>();
        if (members != null && members.get("whites") != null) {
            whites = members.get("whites");
        }
        if (members != null && members.get("blacks") != null) {
            blacks = members.get("blacks");
        }
        List<String> uids = new ArrayList<>();
        for (String uid : list) {
            if (blacks != null && blacks.contains(uid)) {
                // 黑名单用户
                redis_incr(key_black);
                continue;
            }
            if (whites != null && whites.contains(uid)) {
                continue;
            }
            uids.add(uid);
        }
        if (whites != null && whites.size() != 0) {
            uids.addAll(whites);
        }
        return uids;
    }

    /**
     * 新增sftp文件同步记录
     *
     * @param groupId
     * @param strategyId
     */
    private String create_sftp_data_sync_record(String groupId, String strategyId) {
        String today = DateUtil.getDateToday();
        Map<String, Object> map = new HashMap<>();
        map.put("groupId", groupId);
        map.put("strategyId", strategyId);
        map.put("localPath", "/nfs_shared/fine-operation/message-send-record-label/");
        map.put("fileName", "a_10008_80031_" + today + "_" + strategyId + ".dat");
        messageNoticeTaskMapper.create_sftp_data_sync_record(map);
        return map.get("localPath").toString() + map.get("fileName").toString();
    }

    /**
     * 原子自增
     *
     * @return
     */
    private void redis_incr(String key_name) {
        ValueOperations<Serializable, Object> operations = redisTemplate.opsForValue();
        RedisAtomicLong redisAtomicLong = new RedisAtomicLong(key_name, redisTemplate.getConnectionFactory());
        redisAtomicLong.getAndIncrement();
    }

}
